package com.xinqi.modules.file.mq.consumer;

import com.aliyun.oss.ClientException;
import com.xinqi.common.core.enums.RocketMqEnum;
import com.xinqi.common.third.oss.slice.config.OssSliceConfig;
import com.xinqi.modules.file.course.common.CommonService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.RPCHook;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * 示范课程消费消息
 *
 * @author dzy
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class OssSliceConsumerListener {

    private final OssSliceConfig ossSliceConfig;

    private final CommonService commonService;

    /**
     * 替换为您阿里云账号的AccessKey ID和AccessKey Secret。
     * 请确保环境变量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已设置。
     */
    private RPCHook getAclRPCHook() {
        return new AclClientRPCHook(new SessionCredentials(ossSliceConfig.getAccessKey(), ossSliceConfig.getScreenKey()));
    }

    @Async
    public void pushConsumerMsg(RocketMqEnum mqEnum) throws ClientException {
        log.info("OSS切片--MQ监听启动 => {}", mqEnum);
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(mqEnum.getCode(), getAclRPCHook(), new AllocateMessageQueueAveragely(), true, null);
        //设置为阿里云消息队列RocketMQ版实例的接入点。
        consumer.setNamesrvAddr(ossSliceConfig.getMqEndpoint());
        //阿里云上消息轨迹需要设置为CLOUD方式，在使用云上消息轨迹的时候，需要设置此项，如果不开启消息轨迹功能，则运行不设置此项。
        consumer.setAccessChannel(AccessChannel.CLOUD);
        try {
            this.startConsumer(consumer);
        } catch (Exception e) {
            this.startConsumer(consumer);
        }

    }

    /**
     * Rocket消费任务开启
     */
    private void startConsumer(DefaultMQPushConsumer consumer) {
        try {
            // 设置为您在阿里云消息队列RocketMQ版控制台上创建的Topic。
            consumer.subscribe(ossSliceConfig.getMqTopic(), "*");
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext context) {
                    log.info("监听到新的消息=> {}", msgs.size());
                    if (commonService.commonAliSlice(msgs)) {
                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                    } else {
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
            });
            consumer.start();
        } catch (Exception e) {

        }


    }



}
